package com.leansoft.bigqueue.page;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.leansoft.bigqueue.cache.ILRUCache;
import com.leansoft.bigqueue.cache.LRUCacheImpl;
import com.leansoft.bigqueue.utils.FileUtil;

import static java.nio.channels.FileChannel.MapMode.READ_WRITE;

/**
 * Mapped mapped page resource manager,
 * responsible for the creation, cache, recycle of the mapped pages.
 * <p>
 * automatic paging and swapping algorithm is leveraged to ensure fast page fetch while
 * keep memory usage efficient at the same time.
 *
 * @author bulldog
 */
public class MappedPageFactoryImpl implements IMappedPageFactory {

    private final static Logger logger = LoggerFactory.getLogger(MappedPageFactoryImpl.class);

    private int pageSize; // 页大小
    private String pageDir; // 页目录
    private File pageDirFile; // 页目录
    private String pageFile; // 页文件
    private long ttl;  // 缓存过期时间

    private final Object mapLock = new Object();
    private final Map<Long, Object> pageCreationLockMap = new HashMap<Long, Object>(); // 每一个页一个锁

    public static final String PAGE_FILE_NAME = "page"; // page 文件名称
    public static final String PAGE_FILE_SUFFIX = ".dat"; // page 文件后缀

    private ILRUCache<Long, MappedPageImpl> cache; // 缓存

    public MappedPageFactoryImpl(int pageSize, String pageDir, long cacheTTL) {
        this.pageSize = pageSize;
        this.pageDir = pageDir;
        this.ttl = cacheTTL; // 缓存过期时间
        this.pageDirFile = new File(this.pageDir);
        if (!pageDirFile.exists()) {
            pageDirFile.mkdirs();
        }
        if (!this.pageDir.endsWith(File.separator)) {
            this.pageDir += File.separator;
        }
        this.pageFile = this.pageDir + PAGE_FILE_NAME + "-"; // 页文件
        this.cache = new LRUCacheImpl<Long, MappedPageImpl>();
    }

    public IMappedPage acquirePage(long index) throws IOException {
        MappedPageImpl mpi = cache.get(index);
        if (mpi == null) { // not in cache, need to create one
            try {
                Object lock = null;
                synchronized (mapLock) { // 锁住 map
                    // 如果该 index 对应的页没有锁，则新建一个 lock
                    if (!pageCreationLockMap.containsKey(index)) {
                        pageCreationLockMap.put(index, new Object());
                    }
                    lock = pageCreationLockMap.get(index);
                }
                // 现在只锁这个页
                synchronized (lock) { // only lock the creation of page index
                    mpi = cache.get(index); // double check 再次检查缓存，防止缓存失效
                    if (mpi == null) {
                        RandomAccessFile raf = null;
                        FileChannel channel = null;
                        try {
                            // 返回当前索引对应的页文件名称
                            String fileName = this.getFileNameByIndex(index);
                            raf = new RandomAccessFile(fileName, "rw"); // 可读写打开文件
                            channel = raf.getChannel(); // 拿到文件 channel
                            // 将文件映射到可读写的内存 buffer 中
                            MappedByteBuffer mbb = channel.map(READ_WRITE, 0, this.pageSize);
                            mpi = new MappedPageImpl(mbb, fileName, index); // 新建页
                            cache.put(index, mpi, ttl); // 将页加入缓存
                            if (logger.isDebugEnabled()) {
                                logger.debug("Mapped page for " + fileName + " was just created and cached.");
                            }
                        } finally {
                            if (channel != null) channel.close(); // 关闭 channel
                            if (raf != null) raf.close(); // 关闭 file
                        }
                    }
                }
            } finally {
                synchronized (mapLock) { // 加锁 map
                    pageCreationLockMap.remove(index); // 移除页锁
                }
            }
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("Hit mapped page " + mpi.getPageFile() + " in cache.");
            }
        }

        return mpi;
    }

    private String getFileNameByIndex(long index) {
        return this.pageFile + index + PAGE_FILE_SUFFIX;
    }


    public int getPageSize() {
        return pageSize;
    }

    public String getPageDir() {
        return pageDir;
    }

    public void releasePage(long index) {
        cache.release(index); // 释放页
    }

    /**
     * thread unsafe, caller need synchronization
     */
    @Override
    public void releaseCachedPages() throws IOException {
        cache.removeAll();
    }

    /**
     * thread unsafe, caller need synchronization
     */
    @Override
    public void deleteAllPages() throws IOException {
        cache.removeAll();
        Set<Long> indexSet = getExistingBackFileIndexSet();
        this.deletePages(indexSet); // 删除页
        if (logger.isDebugEnabled()) {
            logger.debug("All page files in dir " + this.pageDir + " have been deleted.");
        }
    }

    /**
     * thread unsafe, caller need synchronization
     */
    @Override
    public void deletePages(Set<Long> indexes) throws IOException {
        if (indexes == null) return;
        for (long index : indexes) {
            this.deletePage(index);
        }
    }

    /**
     * thread unsafe, caller need synchronization
     */
    @Override
    public void deletePage(long index) throws IOException {
        // remove the page from cache first
        cache.remove(index);
        String fileName = this.getFileNameByIndex(index);
        int count = 0;
        int maxRound = 10; // 最大重试次数
        boolean deleted = false;
        while (count < maxRound) {
            try {
                FileUtil.deleteFile(new File(fileName));
                deleted = true;
                break;
            } catch (IllegalStateException ex) {
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                } // 如果删除失败，那么 sleep 200ms 再删
                count++;
                if (logger.isDebugEnabled()) {
                    logger.warn("fail to delete file " + fileName + ", tried round = " + count);
                }
            }
        }
        if (deleted) {
            logger.info("Page file " + fileName + " was just deleted.");
        } else {
            logger.warn("fail to delete file " + fileName + " after max " + maxRound + " rounds of try, you may delete it manually.");
        }
    }

    @Override
    public Set<Long> getPageIndexSetBefore(long timestamp) {
        Set<Long> beforeIndexSet = new HashSet<Long>();
        File[] pageFiles = this.pageDirFile.listFiles();
        if (pageFiles != null && pageFiles.length > 0) {
            for (File pageFile : pageFiles) {
                // 如果文件的 最后修改时间 小于 当前时间戳
                if (pageFile.lastModified() < timestamp) {
                    String fileName = pageFile.getName();
                    if (fileName.endsWith(PAGE_FILE_SUFFIX)) { // 是 page 文件
                        long index = this.getIndexByFileName(fileName);
                        beforeIndexSet.add(index); // 那么加入
                    }
                }
            }
        }
        return beforeIndexSet;
    }

    // 解析文件名称，得到索引
    private long getIndexByFileName(String fileName) {
        int beginIndex = fileName.lastIndexOf('-');
        beginIndex += 1;
        int endIndex = fileName.lastIndexOf(PAGE_FILE_SUFFIX);
        String sIndex = fileName.substring(beginIndex, endIndex);
        long index = Long.parseLong(sIndex);
        return index;
    }

    /**
     * thread unsafe, caller need synchronization
     *
     * @throws IOException
     */
    @Override
    public void deletePagesBefore(long timestamp) throws IOException {
        Set<Long> indexSet = this.getPageIndexSetBefore(timestamp); // 得到时间戳之前的修改文件
        this.deletePages(indexSet); // 删除
        if (logger.isDebugEnabled()) {
            logger.debug("All page files in dir [" + this.pageDir + "], before [" + timestamp + "] have been deleted.");
        }
    }

    @Override
    public void deletePagesBeforePageIndex(long pageIndex) throws IOException {
        Set<Long> indexSet = this.getExistingBackFileIndexSet(); // 得到所有文件
        for (Long index : indexSet) {
            if (index < pageIndex) { // 如果文件索引小于 pageIndex 那么删除
                this.deletePage(index);
            }
        }
    }


    @Override
    public Set<Long> getExistingBackFileIndexSet() {
        Set<Long> indexSet = new HashSet<Long>();
        File[] pageFiles = this.pageDirFile.listFiles(); // 获得文件夹下的所有文件
        if (pageFiles != null && pageFiles.length > 0) {
            for (File pageFile : pageFiles) {
                String fileName = pageFile.getName();
                if (fileName.endsWith(PAGE_FILE_SUFFIX)) { // 如果以 .dat 结尾的就是 backfile
                    long index = this.getIndexByFileName(fileName); // 获得文件索引
                    indexSet.add(index); // 加入
                }
            }
        }
        return indexSet;
    }

    @Override
    public int getCacheSize() {
        return cache.size();
    }

    // for testing 有多少文件被锁了
    int getLockMapSize() {
        return this.pageCreationLockMap.size();
    }

    // 得到该索引文件的最后修改时间
    @Override
    public long getPageFileLastModifiedTime(long index) {
        String pageFileName = this.getFileNameByIndex(index);
        File pageFile = new File(pageFileName);
        if (!pageFile.exists()) {
            return -1L;
        }
        return pageFile.lastModified();
    }

    @Override
    public long getFirstPageIndexBefore(long timestamp) {
        Set<Long> beforeIndexSet = getPageIndexSetBefore(timestamp);
        if (beforeIndexSet.size() == 0) return -1L;
        // 直接树化，然后找到第一页
        TreeSet<Long> sortedIndexSet = new TreeSet<Long>(beforeIndexSet);
        Long largestIndex = sortedIndexSet.last();
        return largestIndex;
    }

    /**
     * thread unsafe, caller need synchronization
     */
    @Override
    public void flush() {
        Collection<MappedPageImpl> cachedPages = cache.getValues();
        for (IMappedPage mappedPage : cachedPages) {
            mappedPage.flush();
        }
    }

    // 获得所有的 back file
    @Override
    public Set<String> getBackPageFileSet() {
        Set<String> fileSet = new HashSet<String>();
        File[] pageFiles = this.pageDirFile.listFiles();
        if (pageFiles != null && pageFiles.length > 0) {
            for (File pageFile : pageFiles) {
                String fileName = pageFile.getName();
                if (fileName.endsWith(PAGE_FILE_SUFFIX)) {
                    fileSet.add(fileName);
                }
            }
        }
        return fileSet;
    }

    // 得到所有 back file 的大小
    @Override
    public long getBackPageFileSize() {
        long totalSize = 0L;
        File[] pageFiles = this.pageDirFile.listFiles();
        if (pageFiles != null && pageFiles.length > 0) {
            for (File pageFile : pageFiles) {
                String fileName = pageFile.getName();
                if (fileName.endsWith(PAGE_FILE_SUFFIX)) {
                    totalSize += pageFile.length();
                }
            }
        }
        return totalSize;
    }

}
